package biz.datalk.industrialland.admin.modules.redisdelay;

import biz.datalk.commons.utils.BasicParserUtils;
import biz.datalk.commons.utils.IdProvider;
import biz.datalk.framework.strategy.Strategy;
import biz.datalk.framework.strategy.StrategyHandlerContext;
import biz.datalk.industrialland.admin.config.ThreadPoolExecutorConfig;
import biz.datalk.industrialland.admin.config.properties.RedisDelayTaskExecutorProperties;
import biz.datalk.industrialland.admin.modules.redisdelay.stragety.Ds;
import biz.datalk.industrialland.admin.modules.redisdelay.stragety.DsDTOWrapper;
import biz.datalk.industrialland.admin.modules.redisdelay.stragety.DsVOWrapper;
import biz.datalk.industrialland.common.component.SemaphoreReleaseOnlyOne;
import biz.datalk.industrialland.common.component.filter.MDCFilter;
import biz.datalk.industrialland.common.def.CommonConstant;
import biz.datalk.industrialland.common.exception.ApplicationException;
import biz.datalk.industrialland.common.util.RedisUtil;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.MDC;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.env.Environment;
import org.springframework.core.io.FileSystemResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.io.File;
import java.util.*;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author tarofang@163.com
 * @date 2024-01-18
 */
@Slf4j
@Component
public class RedisDelayTaskContext implements InitializingBean, DisposableBean {
    public static final String DELAY_SCHEDULE_REDIS_KEY = CommonConstant.DELAY_SCHEDULE_REDIS_KEY;
    private boolean _init;
    private final AtomicBoolean _destroy = new AtomicBoolean(Boolean.FALSE);
    private final AtomicInteger atomicInteger = new AtomicInteger(0);
    private File _luaFile;

    public static final String LUA_NAME = "zsetOffer.lua";

    private final RedisUtil redisUtil;
    private final Environment environment;
    private final ThreadPoolTaskExecutor redisDelayTaskThreadPoolTaskExecutor;
    private final Semaphore redisDelayTaskSemaphore;
    private final RedisDelayTaskExecutorProperties redisDelayTaskExecutorProperties;
    private final StrategyHandlerContext strategyHandlerContext;

    public RedisDelayTaskContext(
            RedisUtil redisUtil, //
            Environment environment, //
            @Autowired(required = false) @Qualifier(ThreadPoolExecutorConfig.EXECUTOR_BEAN_NAME_REDIS_DELAY_TASK) ThreadPoolTaskExecutor redisDelayTaskThreadPoolTaskExecutor, //
            @Autowired(required = false) @Qualifier(ThreadPoolExecutorConfig.SEMAPHORE_BEAN_NAME_REDIS_DELAY_TASK) Semaphore redisDelayTaskSemaphore, //
            RedisDelayTaskExecutorProperties redisDelayTaskExecutorProperties, //
            StrategyHandlerContext strategyHandlerContext) {
        this.redisUtil = redisUtil;
        this.environment = environment;
        this.redisDelayTaskThreadPoolTaskExecutor = redisDelayTaskThreadPoolTaskExecutor;
        this.redisDelayTaskSemaphore = redisDelayTaskSemaphore;
        this.redisDelayTaskExecutorProperties = redisDelayTaskExecutorProperties;
        this.strategyHandlerContext = strategyHandlerContext;
    }

    @Value("${delay_schedule_retry_gap_millisecond:300000}") // 默认 5 分钟
    private long delayScheduleRetryGapMillisecond;
    @Value("${delay_schedule_semaphore_acquire_timeout_millisecond:300000}") // 默认 5 分钟
    private long delayScheduleSemaphoreAcquireTimeoutMillisecond;

    @Override
    public void afterPropertiesSet() throws Exception {
        // Runtime.getRuntime().addShutdownHook(new Thread(() -> {
        //     _destroy.set(Boolean.TRUE);
        // }));
        this.init();
    }

    @Override
    public void destroy() throws Exception {
        _destroy.set(Boolean.TRUE);
    }

    private void init() throws Exception {
        if (_init) {
            return;
        }

        String path = environment.getProperty("datalk-delay-redis-lua", String.class, StringUtils.EMPTY);
        log.info("Configuration file:[path={}]", path);
        File luaFile = new File(path);
        if (!luaFile.exists()) {
            path = RedisDelayTaskContext.class.getResource(LUA_NAME).getPath();
            log.info("Class resource:[path={}]", path);
            luaFile = new File(path);
        }
        if (!luaFile.exists()) {
            throw new ApplicationException("lua file no exits.");
        }
        this._luaFile = luaFile;
        _init = true;

        Thread thread = new Thread(this::offer);
        thread.setName("Redis-Delay-Task-Context-Thread");
        thread.start();
    }

    @SuppressWarnings("rawtypes")
    private void offer() {
        try {

            FileSystemResource resource = new FileSystemResource(_luaFile);
            // FileUrlResource resource = new FileUrlResource("file:" + path);
            // InputStreamResource resource = new InputStreamResource(Files.newInputStream(luaFile.toPath()));
            DefaultRedisScript<ArrayList> redisScript = new DefaultRedisScript<>();
            redisScript.setLocation(resource);
            redisScript.setResultType(ArrayList.class);

            String key = DELAY_SCHEDULE_REDIS_KEY;
            long minScore = 0L;
            long offset = 0L;
            long limit = 1;
            List<String> keys = Collections.singletonList(key);
            RedisTemplate<String, Object> redisTemplate = redisUtil.getRedisTemplate();
            int maxIdleTimes = 300;
            int maxIdleEndTimes = Integer.MAX_VALUE - 10000;
            int maxIdleTimes10W_5 = 100000;
            while (true) {
                if (_destroy.get()) {
                    log.warn("The RedisDelayTaskContext destroyed.");
                    break;
                }
                int atomicInt = atomicInteger.get();
                // log.info("atomicInteger.get()={}", atomicInt);
                if (atomicInt > maxIdleTimes) {
                    try {
                        Thread.sleep(1000L);
                    } catch (Exception ex) {
                        log.error("{}", ex.getMessage(), ex);
                    }
                }
                if (atomicInt > maxIdleTimes10W_5) {
                    try {
                        Thread.sleep(5000L);
                    } catch (Exception ex) {
                        log.error("{}", ex.getMessage(), ex);
                    }
                }

                // 防止 redis 报错导致 跳出循环
                ArrayList result = null;
                SemaphoreReleaseOnlyOne releaseOnlyOne = null;
                try {
                    // 该语句可能造成 停服时，一直等待
                    if (redisDelayTaskExecutorProperties.getEnable()) {
                        releaseOnlyOne = new SemaphoreReleaseOnlyOne(redisDelayTaskSemaphore);
                        if (releaseOnlyOne.tryAcquire(delayScheduleSemaphoreAcquireTimeoutMillisecond, TimeUnit.MILLISECONDS)) {
                            long maxScore = System.currentTimeMillis() + 1000L;
                            result = redisTemplate.execute(redisScript, keys, minScore, maxScore, offset, limit);
                        }
                    } else {
                        long maxScore = System.currentTimeMillis() + 1000L;
                        result = redisTemplate.execute(redisScript, keys, minScore, maxScore, offset, limit);
                    }
                } catch (Exception ex) {
                    log.warn("Offer redis error:{}", ex.getMessage());
                }
                int size = result == null ? 0 : result.size();
                if (log.isTraceEnabled()) {
                    log.trace("resultSize={}, atomicInt={}", size, atomicInt);
                }
                if (size != 2) {
                    int incr = atomicInteger.incrementAndGet();
                    if (incr > maxIdleEndTimes) {
                        atomicInteger.set(maxIdleTimes + 100);
                    }
                    if (redisDelayTaskExecutorProperties.getEnable()) {
                        releaseOnlyOne.release();
                    }
                    continue;
                }
                atomicInteger.set(0);
                Object memberObj = result.get(0);
                Object scoreObj = result.get(1);
                // log.info("memberObjClass={}, scoreObjClass={}", memberObj.getClass().getName(), scoreObj.getClass().getName());
                String member = String.valueOf(memberObj);
                long score = BasicParserUtils.parseLong(String.valueOf(scoreObj), 0L);
                if (redisDelayTaskExecutorProperties.getEnable()) {
                    this.doBusinessHandler(key, member, score, releaseOnlyOne);
                } else {
                    this.doBusiness(key, member, score, releaseOnlyOne);
                }
            } // end of while (true) {
        } catch (Exception ex) {
            log.error("Offer redis delay task error:{}", ex.getMessage(), ex);
        }
    }

    private void doBusinessHandler(String key, String member, long score, final SemaphoreReleaseOnlyOne semaphoreReleaseOnlyOne) {
        try {
            redisDelayTaskThreadPoolTaskExecutor.execute(() -> this.doBusiness(key, member, score, semaphoreReleaseOnlyOne));
        } catch (Exception ex) {
            log.error("{}", ex.getMessage(), ex);
        }
    }

    private void doBusiness(String key, String member, long score, final SemaphoreReleaseOnlyOne semaphoreReleaseOnlyOne) {
        try {
            if (!redisDelayTaskExecutorProperties.getEnable()) {
                setMdcContext();
            }
            if (log.isTraceEnabled()) {
                log.trace("doBusiness: key={}, member={}, score={}", key, member, score);
            }
            // parkingLotInOutPushHandler.doBusiness(key, member, score);

            DsDTOWrapper dto = new DsDTOWrapper();
            dto.setKey(key);
            dto.setMember(member);
            dto.setScore(score);
            Strategy<DsDTOWrapper, DsVOWrapper> startWithHandlerInstanceStartWith = strategyHandlerContext.getHandlerInstanceStartWith(member, Ds.class, DsDTOWrapper.class, DsVOWrapper.class);
            if (startWithHandlerInstanceStartWith == null) {
                log.warn("Invalid member, startWithHandlerInstanceStartWith is null. [key={}, member={}, score={}]", key, member, score);
                return;
            }
            DsVOWrapper vo = startWithHandlerInstanceStartWith.handler(dto);
            if (vo == null) {
                log.warn("DsVOWrapper is null. [key={}, member={}, score={}]", key, member, score);
                return;
            }
            // 执行成功， 打印结果
            if (vo.isSuccess()) {
                if (log.isInfoEnabled()) {
                    log.info("Exec success. [key={}, member={}, score={}]", key, member, score);
                }
                return;
            }
            // 执行失败， 重新进入队列， 5 分钟后再次执行
            reEnterRedis(key, member);
        } catch (Exception ex) {
            log.error("{}", ex.getMessage(), ex);
            reEnterRedis(key, member);
        } finally {
            if (semaphoreReleaseOnlyOne != null) {
                semaphoreReleaseOnlyOne.release();
            }
        }
    }

    private void reEnterRedis(String key, String member) {
        try {
            // 5 分钟后再次执行
            redisUtil.getRedisTemplate().opsForZSet().add(DELAY_SCHEDULE_REDIS_KEY, member, System.currentTimeMillis() + delayScheduleRetryGapMillisecond);
            log.info("Re-enter the zset. [key={}, member={}]", key, member);
        } catch (Exception neEx) {
            log.error("{}", neEx.getMessage(), neEx);
        }
    }

    private void setMdcContext() {
        Map<String, String> previous = MDC.getCopyOfContextMap();
        Map<String, String> map = Optional.ofNullable(previous).orElse(Maps.newHashMap());
        map.put(MDCFilter.TRACE_ID, IdProvider.createUUIDId()); // 每次都生成新的 traceId
        MDC.setContextMap(map);
    }

}
